package org.xukai.remoting.io.nio.weiyi;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @ Type : com.weiyi.wave.core.service
 * @ Author : TimLi
 * @ Desc :
 * @ Date : 2016-10-10 19:04
 */
public class SocketSelectorTask implements  Runnable {

    private static final Logger logger = LoggerFactory.getLogger(SocketSelectorTask.class);

    private final Selector selector;

    private Map<String, SocketChannel> clientsMap = new HashMap<String, SocketChannel>();

    public SocketSelectorTask(Selector selector)
    {
        this.selector = selector;
    }


    @Override
    public void run() {
        ExecutorService executorService = Executors.newCachedThreadPool();
        while (true){
            try {
                int connects = selector.select(2000);// 每次最多阻塞2秒
                if (connects == 0) {
                    System.out.println("没有请求...");
                    continue;
                } else {
                    System.out.println("请求来了...");
                }
                Iterator<SelectionKey> it = selector.selectedKeys().iterator();
                while (it.hasNext()) {
                    SelectionKey selKey = it.next();
                    it.remove();
                    if (selKey.isAcceptable()) {
                        accept(selKey);
                    } else if(selKey.isReadable()){
                        read(selKey);
                    }
                }
            } catch (IOException e) {
                logger.error("socket selector异常",e);
            }finally {
          //      executorService.shutdown();
            }
        }
    }
    private void accept(SelectionKey key) {
        try {
//            //1 获取服务通道
            ServerSocketChannel ssc =  (ServerSocketChannel) key.channel();
            //2 执行阻塞方法
            SocketChannel sc = ssc.accept();
            //3 设置阻塞模式
            sc.configureBlocking(false);
            clientsMap.put(sc.getRemoteAddress().toString(), sc);
            //4 注册到多路复用器上，并设置读取标识
            SelectionKey key1 = sc.register(this.selector, SelectionKey.OP_READ);
//            if (null == key1) {
//                SelectionKey key1 = sc.register(this.selector, SelectionKey.OP_READ);
//            }
//            if (null == key2) {
//                SelectionKey key2 = sc.register(this.selector, SelectionKey.OP_WRITE);
//            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    private void read(SelectionKey key){
        SocketChannel channel = (SocketChannel)key.channel();

        try {
//            Integer bytesRead = 0;
            ArrayList<Integer> list = new ArrayList<>();
            ByteBuffer by = ByteBuffer.allocate(16);
            boolean caseFlag=false;//判断返回包头是否是病例
            int bytesRead = 0;
            while (true) {
                bytesRead = channel.read(by);
                if (bytesRead <= 0) {
                    break;
                }
                WanPacketHeader header = StringTranUtils.ByteToHeader(by.array());
                if (header.getnPldLen() < 0 || header.getnPldType() < 0) {
                    logger.error("获取header信息失败");
                    return;
                }
                by.flip();
                byte[] bytes1 = new byte[by.remaining()];
                by.get(bytes1);
//                send(channel,by);
                by.clear();
                by = ByteBuffer.allocate(header.getnPldLen());
                bytesRead = channel.read(by);
                if (bytesRead <= 0) {
                    break;
                }
                by.flip();
                byte[] bytes2 = new byte[by.remaining()];
                by.get(bytes2);
                byte[] bytes = byteMerger(bytes1, bytes2);
                ByteBuffer send = ByteBuffer.allocate(bytes.length);
                send.put(bytes);
                send(channel,send);
                by.clear();
                by = ByteBuffer.allocate(16);
            }
//            while (true) {
//                bytesRead = channel.read(buffer);
//                if (bytesRead <= 0) {
//                    break;
//                }
//                buffer.flip();
//                list.add(bytesRead);
//                send(channel,buffer);
//            }
            channel.register(selector,SelectionKey.OP_READ);
//            System.out.println("list" + list.size());
        } catch (Exception e) {
            logger.error("采集患者电生理信息异常", e);
        } finally {
            logger.info("一次读取结束了");
        }
    }

    private void send(SocketChannel channel,ByteBuffer buffer) throws IOException {
        if(!clientsMap.isEmpty()){

            for(Map.Entry<String, SocketChannel> entry : clientsMap.entrySet()){
                buffer.flip();
                SocketChannel temp = entry.getValue();
                if(!channel.equals(temp)){
                    //输出到通道
                    temp.write(buffer);
                }
            }
        }
        Socket s = channel.socket();
        SocketAddress remoteSocketAddress = s.getRemoteSocketAddress();
//        clientsMap.put(remoteSocketAddress.toString(), channel);
    }
//    private void write(SelectionKey key){
//        SocketChannel sc = (SocketChannel) key.channel();
//        by.flip();
//        try {
//            sc.write(by);
//            by.clear();
//            sc.register(selector,SelectionKey.OP_READ);
//        } catch (IOException e) {
//            e.printStackTrace();
//        }
//    }

    public static void main(String[] args) {
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        try {
            logger.info("Socket服务开始启动.......... port:"+8072);
            Selector selector = Selector.open();
            ServerSocketChannel server = ServerSocketChannel.open();
            server.configureBlocking(false);
            server.socket().bind(new InetSocketAddress(Integer.parseInt("8072")), 5);
            SelectionKey key1 = server.register(selector, SelectionKey.OP_ACCEPT);
            executorService.submit(new SocketSelectorTask(selector));
            logger.info("Socket服务开始启动成功................");
        } catch (IOException e) {
            logger.error("Socket服务启动失败....", e);
        }finally {
            executorService.shutdown();
        }
    }
    //java 合并两个byte数组
    private byte[] byteMerger(byte[] byte_1, byte[] byte_2){
        byte[] byte_3 = new byte[byte_1.length+byte_2.length];
        System.arraycopy(byte_1, 0, byte_3, 0, byte_1.length);
        System.arraycopy(byte_2, 0, byte_3, byte_1.length, byte_2.length);
        return byte_3;
    }
}
